#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：sumo_jingtai
@File ：离线数据发送.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/6/13 8:41
'''
import json
import time
from datetime import datetime

from kafka import KafkaProducer
from loguru import logger

from utils.config_temp import KAFKA_HOST,RADAR_DATA_LIST_CJ

station_orgCode_dict = {11: "e1_data_CJ-10-01"}

kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                               value_serializer=lambda x: json.dumps(x).encode('utf-8'))
topic_list = ['e1_data_CJ-10-01-0', 'e1_data_CJ-10-02-0', 'e1_data_JJ-10-01-0', 'e1_data_JJ-10-02-0']


def get_timestamp_milliseconds(t_str):
    """
    将ISO 8601时间戳字符串转为13位毫秒级时间戳
    :param t_str: ISO 8601时间戳字符串
    :return: 13位毫秒级时间戳
    """
    # t_str = data['timestamp']
    # 解析 ISO 8601 时间字符串为 datetime 对象
    dt = datetime.strptime(t_str, "%Y-%m-%d %H:%M:%S:%f")
    # 将 datetime 对象转换为秒级时间戳
    timestamp_seconds = dt.timestamp()
    # 将秒级时间戳转换为毫秒级时间戳
    timestamp_milliseconds = int(timestamp_seconds * 1000)
    return timestamp_milliseconds


def send_by_sn():
    file_path = r"C:\Users\xinyingjie\Downloads\e1_source_5000.log"
    before_time = None
    with open(file_path, 'r', encoding='utf-8') as f:
        while True:
            line = f.readline().strip()
            if line:
                target_info = {}
                for sn,sn_info in RADAR_DATA_LIST_CJ.items():
                    target_info[sn] = []
                data = json.loads(line)
                # orgCode = data['orgCode']
                targetList = data['targetList']
                for target in targetList:
                    target['station'] = target['station']
                    station = target['station']
                    target['orgCode'] = RADAR_DATA_LIST_CJ[station]['orgCode']
                    target_info[station].append(target)
                for sn,t_info in target_info.items():
                    if len(t_info) > 0:
                        data['orgCode'] = RADAR_DATA_LIST_CJ[sn]['orgCode']
                        data['targetList'] = t_info
                        if sn <= 18:
                            logger.info(f"CJ-10-SUM {data}")
                            kafka_producer.send("CJ-10-SUM",data,key=str(time.time()))
                        else:
                            logger.info(f"JJ-10-SUM {data}")
                            kafka_producer.send("JJ-10-SUM", data, key=str(time.time()))
                globalTime = data['globalTime']
                if before_time:
                    sleep_time = abs(before_time - globalTime) / 1000
                    time.sleep(sleep_time)
                before_time = globalTime
            else:
                break


send_by_sn()
